package com.minelx.datamicroscope.core.query;

import com.minelx.datamicroscope.core.values.CommonField;
import com.minelx.hellopandas.core.DataFrame;
import com.minelx.hellopandas.core.DataFrameDescription;
import com.minelx.hellopandas.core.Pandas;
import com.minelx.hellopandas.core.PandasMethodProvider;
import io.gitee.minelx.filesystem.Directory;
import io.gitee.minelx.filesystem.TempPath;
import com.minelx.sqlstructures.core.ColumnType;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import static java.util.Collections.emptyList;
import static java.util.Collections.singletonList;
import static java.util.stream.Collectors.toList;
import static java.util.stream.Collectors.toMap;

public class Resource implements AutoCloseable {

    private final Directory storedCachedIdentitiesDirectory;

    private final DataFrame dataFrame;

    private final Map<String, DataFrame> cachedDataFrames;

    private final List<TempPath> cachedIdentities;

    private final List<String> containedNames;

    private final List<String> clueNames;

    private final PandasMethodProvider pandasMethodProvider;

    private Resource(Directory storedCachedIdentitiesDirectory,
                     PandasMethodProvider pandasMethodProvider,
                     
                     DataFrame dataFrame,
                     List<String> clueNames,
                     List<String> containedNames) {
        this.storedCachedIdentitiesDirectory = storedCachedIdentitiesDirectory;
        this.dataFrame = dataFrame;
        this.containedNames = containedNames;
        this.clueNames = clueNames;
        this.pandasMethodProvider = pandasMethodProvider;
        cachedDataFrames = new HashMap<>();
        cachedIdentities = new ArrayList<>();
    }

    public DataFrame dataFrame() {
        return dataFrame;
    }

    public List<String> clues() {
        return clueNames;
    }

    public boolean contains(String nickname) {
        return containedNames.contains(nickname);
    }

    public DataFrame identify(String nickname) {
        // nickname should be existed in resource dataFrame
        if (!dataFrame.description().dtypes().containsKey(nickname)) {
            throw new IllegalArgumentException("column is not existed in resource dataFrame: " + nickname);
        }

        // returns cached identity if possible
        if (cachedDataFrames.containsKey(nickname)) {
            return cachedDataFrames.get(nickname);
        }

        // create nickname retained identity file
        TempPath cachedIdentity = TempPath.random(storedCachedIdentitiesDirectory);
        cachedIdentities.add(cachedIdentity);

        DataFrameDescription retained = pandasMethodProvider.retains(dataFrame.description(),
                singletonList(nickname),
                // @identity@ resource obtaining @identity@
                cachedIdentity.file().path().uri());
        DataFrame retainedFrame = Pandas.of(pandasMethodProvider).dataFrame(retained);
        // do identity preprocessing
        retainedFrame.dropNaN(singletonList(nickname), "any");
        retainedFrame.dropDuplicates();

        // returns retainedFrame and save identity frame as cache
        cachedDataFrames.put(nickname, retainedFrame);
        return retainedFrame;
    }

    @Override
    public void close() {
        // delete all cached identities.
        cachedIdentities.forEach(TempPath::close);
    }

    public static Resource create(Directory storedCachedIdentitiesDirectory,
                                  PandasMethodProvider pandasMethodProvider,

                                  String resourceURI,
                                  List<AliasedKey> clueKeys,
                                  List<AliasedKey> containedKeys) {
        Map<String, String> dtypes = dtypesFromAliasedKeys(
                Stream.concat(clueKeys.stream(), containedKeys.stream()).collect(toList()));
	    DataFrame resourceFrame = Pandas.of(pandasMethodProvider).dataFrame(DataFrameDescription.create(resourceURI, dtypes));
        return new Resource(
                storedCachedIdentitiesDirectory,
                pandasMethodProvider,
                resourceFrame,
                clueKeys.stream().map(AliasedKey::getAlias).collect(toList()),
                containedKeys.stream().map(AliasedKey::getAlias).collect(toList()));
    }

    public static Resource root(Directory storedCachedIdentitiesDirectory,
                                PandasMethodProvider pandasMethodProvider,

                                String resourceURI,
                                List<AliasedKey> containedKeys) {
        return create(storedCachedIdentitiesDirectory, pandasMethodProvider, resourceURI, emptyList(), containedKeys);
    }

    public static Map<String, String> dtypesFromAliasedKeys(List<AliasedKey> aliasedKeys) {
        return aliasedKeys.stream()
                .collect(toMap(AliasedKey::getAlias, aliasedKey -> getColumnType(aliasedKey.getKey())));
    }

    private static String getColumnType(CommonField key) {
        return literalColumnType(ColumnType.from(key.structure().getType()));
    }

    public static String literalColumnType(ColumnType columnType) {
        if (columnType == ColumnType.INT) {
            return "int";
        } else if (columnType == ColumnType.TEXT) {
            return "text";
        } else if (columnType == ColumnType.DECIMAL) {
            return "float";
        } else {
            throw new UnsupportedOperationException("unsupported column type: " + columnType);
        }
    }
}
